package com.atguigu.flink.state;

import com.atguigu.flink.function.MyUtil;
import com.atguigu.flink.function.WaterSensorMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by Smexy on 2023/3/1
 *
 *  Raw State： 原始状态，用户定义，由用户自己去维护。
 *                  保证数据的一致性，程序重启后，可以基于重启前计算的结果继续运算，需要手动备份和恢复，麻烦！
 *  Managed State： 管理状态，用户定义，由flink帮用户维护。
 *          维护：
 *                  备份。在Task失败时，自动恢复。
 *
 *                  1。使用Flink提供的Managed State的API进行编程，定义状态
 *                  2。 保证程序能够在失败时，自动恢复(重启)
 *
 *
 *    ------------
 *          flink尝试去恢复，但是被 NoRestartBackoffTimeStrategy 阻止了。
 *    Recovery is suppressed by NoRestartBackoffTimeStrategy
 */
public class Demo1_State
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(2);

        //修改策略，允许重启
        // fixedDelayRestart: 最多重试重启3次，每次尝试间隔1000ms
        //env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3,1000));

        //开启ck(备份)。每隔2s备份一次。 无限重启
        env.enableCheckpointing(2000);

                env
                   .socketTextStream("hadoop103", 8888)
                   .map(new MyMapFunction())
                   .addSink(new SinkFunction<String>()
                   {
                       //如果数据中包含了 x，手动模拟异常
                       @Override
                       public void invoke(String value, Context context) throws Exception {

                           if (value.contains("x")){
                               throw new RuntimeException("出错了...");
                           }
                           System.out.println(value);
                       }
                   });


                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }

    //累积输入的每个字符串
    public static class MyMapFunction implements MapFunction<String,String>, CheckpointedFunction
    {

        //声明一个集合，用于累积数据
        private ListState<String> listState;
        //常量
        private String str = "haha";
        //private List<String> strs ;

        //把状态理解为一个集合就可以

        @Override
        public String map(String value) throws Exception {
            //strs.add(value);
            listState.add(value);
            System.out.println(str);
            return listState.get().toString();
        }

        //快照(备份)状态  由备份周期决定执行的频率
        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            System.out.println("MyMapFunction.snapshotState");
            //把之前的List<String> strs 中的内容，存入状态中
            //listState.addAll(strs);
        }

        //初始化状态。  在Task重启时，从之前的备份中帮你恢复状态。 只在Task创建时，执行一次！
        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("MyMapFunction.initializeState");

            // 从备份中，获取状态存储器，从存储器中取一个ListState。通过ListStateDescriptor描述要取的那个ListState
            listState = context.getOperatorStateStore().getListState(new ListStateDescriptor<>("strs", String.class));
            // 把listState中的数据，恢复到strs 中
            //strs=MyUtil.parseToList(listState.get());
        }
    }
}
